package cn._51doit.live.jobs;

import cn._51doit.live.pojo.DataBean;
import cn._51doit.live.udf.IsNewUserFunctionV3;
import cn._51doit.live.udf.JsonToBeanFunction;
import cn._51doit.live.utils.Constants;
import cn._51doit.live.utils.FlinkUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;


/**
 * 使用DataStream API多维度统计新用户
 * 数据中没有isNew这个字段，而是必须实时的计算出该字段，将同一个设备ID的数据，进行keyBy，不论是任何时间的访问数据，一定进入到同一个分区中
 *
 * 按照设备ID进行keyBy，可以保证，同一个设备ID的数据，一定进入到同一个分区，并且可以避免数据倾斜，并且准确
 *
 * 存在的问题？
 *   KeyedState中的key会非常多，如果都存在内存中，会非常消耗内存
 *
 * 解决方案（使用RocksDBStateBackend）
 *   可以存储大量状态
 *   并且支持增量checkpoint
 *
 */
public class NewUserCountV3 {


    public static void main(String[] args) throws Exception {

        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);

        DataStream<String> lines = FlinkUtils.createKafkaStream(parameterTool, SimpleStringSchema.class);

        SingleOutputStreamOperator<DataBean> beanStream = lines.process(new JsonToBeanFunction());

        //beanStream.print();
        //过滤数据，只要appLaunch类型
        SingleOutputStreamOperator<DataBean> filtered = beanStream.filter(bean -> Constants.APP_LAUNCH.equals(bean.getEventId()));

        //按照设备ID进行KeyBy
        KeyedStream<DataBean, String> keyedStream = filtered.keyBy(DataBean::getDeviceId);

        SingleOutputStreamOperator<DataBean> res = keyedStream.process(new IsNewUserFunctionV3());

        //然后将数据写入到ClickHouse中做多维统计


        FlinkUtils.env.execute();


    }
}
